package com.pro
import java.util.Properties

import com.`trait`._
import com.utils.{ConfigUtils, DateUtils, KuduUtils}
import org.apache.kudu.client.CreateTableOptions
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.spark.sql.{DataFrame, SparkSession}


object AdNetworkAnalysis extends Process {
  override def process(spark: SparkSession, kuduContext: KuduContext): Unit = {
    val options = Map(
      "kudu.master"->ConfigUtils.KUDU_MASTER,
      "kudu.table"->"ODS_20190822")
    import org.apache.kudu.spark.kudu._
    spark.read.options(options).kudu.createOrReplaceTempView("t_table_info")

    val result: DataFrame = spark.sql(
      """
        |select  networkmannerid,networkmannername, org_num, valid_num, ad_num, bid_num, bid_success_num,
        | bid_success_num/bid_num bid_success_rat,
        | show_num, click_num, click_num/show_num click_rat, media_show_num, media_click_num,
        | consumtion_number, cost_number
        | from ( select networkmannerid,networkmannername,
        | sum(case when requestmode=1 and processnode>=1 then 1 else 0 end) org_num,
        | sum(case when requestmode=1 and processnode>=2 then 1 else 0 end) valid_num,
        | sum(case when requestmode=1 and processnode=3 then 1 else 0 end) ad_num,
        | sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and isbid=1 and adorderid!=0 then 1 else 0 end) bid_num,
        | sum(case when iseffective=1 and iswin=1 and isbilling=1 and adplatformproviderid>=100000 then 1 else 0 end) bid_success_num,
        | sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) show_num,
        | sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) click_num,
        | sum(case when requestmode=2 and iseffective=1 and isbilling=1 then 1 else 0 end) media_show_num,
        | sum(case when requestmode=3 and iseffective=1 and isbilling=1 then 1 else 0 end) media_click_num,
        | sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and iswin=1 and adorderid>200000 and adcreativeid>200000 then winprice/1000 else 0 end) consumtion_number,
        | sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and iswin=1 and adorderid>200000 and adcreativeid>200000 then adpayment/1000 else 0 end) cost_number
        | from t_table_info group by networkmannerid,networkmannername ) a
      """.stripMargin)

    val columns = Seq("networkmannerid","networkmannername")
    val option = new CreateTableOptions
    import scala.collection.JavaConverters._
    option.addHashPartitions(columns.asJava,3)
    option.setNumReplicas(3)

    KuduUtils.writeToKudu(kuduContext,result.schema,option,s"ad_network_analysis_${DateUtils.getNow()}",columns,result)
    val properties = new Properties()
    properties.setProperty("user","root")
    properties.setProperty("password","123")
    KuduUtils.writeToMysql(ConfigUtils.JDBC_URL,s"ad_network_analysis_${DateUtils.getNow()}",properties,result)
  }
}
